package de.distmlp.preprocessing.normalisation;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.commons.lang3.tuple.ImmutablePair;

import de.distmlp.preprocessing.MLDataProcessor;
import de.distmlp.preprocessing.data.MLDataEntry;

public class Normalizer {
	public static final int THREAD_SIZE = Runtime.getRuntime().availableProcessors();

	/**
	 * The left value is the lower bound. The right value is the upper bound.
	 */
	public static final ImmutablePair<Double, Double> STANDARD_BOUNDARIES = new ImmutablePair<Double, Double>(0d, 1d);

	public static NormalizationValues[] calculateMaxMinValues(final double[][] inputData) {
		if ((inputData == null) || (inputData.length == 0)) {
			return null;
		}

		final ExecutorService threadPool = Executors.newFixedThreadPool(Normalizer.THREAD_SIZE);
		final CompletionService<NormalizationValues[]> pool = new ExecutorCompletionService<NormalizationValues[]>(threadPool);

		final int dataArrayLength = inputData[0].length;
		final int worksize = inputData.length / Normalizer.THREAD_SIZE;
		// den rest übernimmt der letzte Thread.
		final int rest = inputData.length % Normalizer.THREAD_SIZE;
		int numberOfThreads = Normalizer.THREAD_SIZE;

		if (worksize > 0) {
			for (int i = 0; i < numberOfThreads; i++) {
				final int start = i * worksize;
				final int end;
				if (i == (numberOfThreads - 1)) {
					end = ((i + 1) * worksize) + rest;
				} else {
					end = (i + 1) * worksize;
				}
				pool.submit(new Callable<NormalizationValues[]>() {

					@Override
					public NormalizationValues[] call() throws Exception {
						return Normalizer.calculateNormalizationValues(inputData, dataArrayLength, start, end);
					}
				});
			}
		} else {
			numberOfThreads = 1;
			// Data is smaller than number of available threads
			pool.submit(new Callable<NormalizationValues[]>() {

				@Override
				public NormalizationValues[] call() throws Exception {
					return Normalizer.calculateNormalizationValues(inputData, dataArrayLength, 0, rest);
				}
			});
		}
		// Sammeln der Ergebnisse der einzelnen Threads und endgültige
		// Bestimmung von Minima und Maxima.
		final NormalizationValues[] values = new NormalizationValues[dataArrayLength];
		for (int i = 0; i < dataArrayLength; i++) {
			values[i] = new NormalizationValues(Double.MAX_VALUE, Double.MIN_VALUE);
		}
		for (int i = 0; i < numberOfThreads; i++) {
			try {
				final NormalizationValues[] result = pool.take().get();
				for (int j = 0; j < result.length; j++) {
					if (values[j].getMax() < result[j].getMax()) {
						values[j].setMax(result[j].getMax());
					}
					if (values[j].getMin() > result[j].getMin()) {
						values[j].setMin(result[j].getMin());
					}
				}
			} catch (final Exception e) {
				e.printStackTrace();
				return null;
			}
		}
		threadPool.shutdown();
		return values;
	}

	public static double normalizeValue(final NormalizationValues norm, final double input) {
		return Normalizer.normalizeValue(norm, input, Normalizer.STANDARD_BOUNDARIES);
	}

	public static double normalizeValue(final NormalizationValues norm, final double input, final ImmutablePair<Double, Double> boundaries) {
		final double value = (((input - norm.getMin()) / (norm.getMax() - norm.getMin())) * (boundaries.right - boundaries.left))
				+ boundaries.left;
		return value;
	}

	public static double denormalizeValue(final NormalizationValues norm, final double input) {
		return Normalizer.denormalizeValue(norm, input, Normalizer.STANDARD_BOUNDARIES);
	}

	public static double denormalizeValue(final NormalizationValues norm, final double input, final ImmutablePair<Double, Double> boundaries) {
		final double result1 = ((((norm.getMin() - norm.getMax()) * input) - (boundaries.right * norm.getMin())) + (norm.getMax() * boundaries.left))
				/ (boundaries.left - boundaries.right);
		return new BigDecimal(result1).setScale(5, RoundingMode.HALF_DOWN).doubleValue();
	}

	public static List<MLDataEntry> normalizeInput(final List<MLDataEntry> mlData) throws Exception {
		return Normalizer.normalizeInput(mlData, Normalizer.STANDARD_BOUNDARIES);
	}

	/**
	 * Input and output are expected to have the same normalisation boundaries.
	 * 
	 * @param mlData
	 * @param bound
	 * @return
	 * @throws Exception
	 */
	public static List<MLDataEntry> normalizeInput(final List<MLDataEntry> mlData, final ImmutablePair<Double, Double> bound)
			throws Exception {
		if ((mlData == null) || mlData.isEmpty()) {
			return null;
		}
		final int nbInputUnits = mlData.get(0).getFeaturevectorSize();
		final int nbOutputUnits = mlData.get(0).getTargetVectorSize();

		final double[][] input = new double[mlData.size()][];
		final double[][] output = new double[mlData.size()][];
		MLDataProcessor.convertMLDataSetToDouble(mlData, input, output);
		final NormalizationValues[] normValuesInput = Normalizer.calculateMaxMinValues(input);
		final NormalizationValues[] normValuesOutput = Normalizer.calculateMaxMinValues(output);
		Normalizer.normalize2DArray(input, normValuesInput, bound);
		Normalizer.normalize2DArray(output, normValuesOutput, bound);
		final List<MLDataEntry> normMLData = new ArrayList<MLDataEntry>();

		for (int i = 0; i < input.length; i++) {
			final MLDataEntry e = new MLDataEntry(nbInputUnits, nbOutputUnits);
			for (int x = 0; x < input[i].length; x++) {
				e.addInput(x, input[i][x]);
			}
			for (int x = 0; x < output[i].length; x++) {
				e.addOutput(x, output[i][x]);
			}
			normMLData.add(e);
		}
		return normMLData;
	}

	public static void normalize2DArray(final double[][] data, final NormalizationValues[] normValues) throws Exception {
		Normalizer.normalize2DArray(data, normValues, Normalizer.STANDARD_BOUNDARIES);
	}

	/**
	 * Normalizes the input data array.
	 * 
	 * @param data
	 * @param normValues
	 * @throws Exception
	 */
	public static void normalize2DArray(final double[][] data, final NormalizationValues[] normValues,
			final ImmutablePair<Double, Double> boundaries) throws Exception {
		if (data.length == 0) {
			throw new Exception("Array must contain data to normalize!");
		}
		if (data[0].length != normValues.length) {
			throw new Exception("Size of data array and of norm values array must equal!");
		}
		final ExecutorService pool = Executors.newFixedThreadPool(Normalizer.THREAD_SIZE);
		final int worksize = data.length / Normalizer.THREAD_SIZE;
		// den rest übernimmt der letzte Thread.
		final int rest = data.length % Normalizer.THREAD_SIZE;

		if (worksize > 0) {
			for (int i = 0; i < Normalizer.THREAD_SIZE; i++) {
				final int start = i * worksize;
				final int end;
				if (i == (Normalizer.THREAD_SIZE - 1)) {
					end = ((i + 1) * worksize) + rest;
				} else {
					end = (i + 1) * worksize;
				}

				pool.submit(new Runnable() {

					@Override
					public void run() {
						for (int x = start; x < end; x++) {
							final double[] element = data[x];
							for (int i = 0; i < element.length; i++) {
								element[i] = Normalizer.normalizeValue(normValues[i], element[i], boundaries);
							}

						}

					}
				});
			}
		} else {
			// data is smaller than number of available threads
			for (int x = 0; x < rest; x++) {
				final double[] element = data[x];
				for (int i = 0; i < element.length; i++) {
					element[i] = Normalizer.normalizeValue(normValues[i], element[i], boundaries);
				}

			}
		}
		pool.shutdown();
	}

	private static NormalizationValues[] calculateNormalizationValues(final double[][] inputData, final int valueArrayLength,
			final int start, final int end) {
		final NormalizationValues[] values = new NormalizationValues[valueArrayLength];

		for (int i = 0; i < valueArrayLength; i++) {
			values[i] = new NormalizationValues(Double.MAX_VALUE, Double.MIN_VALUE);
		}

		for (int listCounter = start; listCounter < end; listCounter++) {
			for (int valueCounter = 0; valueCounter < valueArrayLength; valueCounter++) {
				if (inputData[listCounter][valueCounter] > values[valueCounter].getMax()) {
					values[valueCounter].setMax(inputData[listCounter][valueCounter]);
				}
				if (inputData[listCounter][valueCounter] < values[valueCounter].getMin()) {
					values[valueCounter].setMin(inputData[listCounter][valueCounter]);
				}
			}
		}
		return values;
	}
}
